package com.gitee.melin.bee.core.jdbc.dialect.postgres;

import com.gitee.melin.bee.core.jdbc.dialect.DefaultTypeConvertor;
import com.gitee.melin.bee.core.jdbc.relational.ConnectionInfo;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.dialect.DataTypeConvertor;
import com.gitee.melin.bee.core.jdbc.dialect.AbstractJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.IDBQuery;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.List;

public class PostgreSqlJdbcDialect extends AbstractJdbcDialect {

    private static final Logger LOG = LoggerFactory.getLogger(PostgreSqlJdbcDialect.class);

    @Override
    public DataSourceType getDataSourceType() {
        return DataSourceType.POSTGRESQL;
    }

    private static final String QUERY_DATABASES_SQL =
            "select datname from pg_catalog.pg_database where datistemplate = false";

    private static final String VERSION_SQL = "SELECT version()";

    private static final double SUPPORT_RELPARTBOUND_START_VERSION = 10;

    private static final String SUPPORT_RELPARTBOUND_STATEMENT = "AND c.relpartbound IS NULL";

    private static final String NOT_SUPPORT_RELPARTBOUND_STATEMENT =  "and not EXISTS(SELECT 1 from pg_inherits  where pg_inherits.inhrelid = c.oid)";

    private Integer isHighVersion = 0;//高版本支持relpartbound字段:1,低于10下版本不支持 -1

    public PostgreSqlJdbcDialect(ConnectionInfo connectionInfo) {
        super(connectionInfo);
    }

    public PostgreSqlJdbcDialect(Connection connection) {
        super(connection);
    }

    @Override
    protected IDBQuery getDBQuery() {
        //判断数据库版本,高版本支持relpartbound字段,低版本不支持
        this.sqlCompatibleVersion();
        if (isHighVersion > 0) {
            return new PostgreSqlQuery(SUPPORT_RELPARTBOUND_STATEMENT);
        }
        return new PostgreSqlQuery(NOT_SUPPORT_RELPARTBOUND_STATEMENT);
    }

    private void sqlCompatibleVersion() {
        if (isHighVersion != 0) {
            return;
        }

        try (Connection connection = this.getConnection();
             Statement statement = connection.createStatement();
             ResultSet versionRs = statement.executeQuery(VERSION_SQL)){
            String version = null;
            while (versionRs.next()) {
                version = versionRs.getString(1);
            }

            if (version.contains("(")) {
                version = version.substring(0, version.indexOf("("));
                version = version.replaceFirst("PostgreSQL", "").trim();
            }

            if (version.contains(".")){
                version = version.split("\\.")[0];
            }
            double versionNumber = Double.parseDouble(version);
            if (versionNumber > SUPPORT_RELPARTBOUND_START_VERSION) {
                isHighVersion = 1;
            } else {
                isHighVersion = -1;
            }
        } catch (Exception ex) {
            isHighVersion = 1;
            LOG.error(ex.getMessage(), ex);
        }
    }

    @Override
    protected DataTypeConvertor getTypeConvert() {
        return new DefaultTypeConvertor();
    }

    @Override
    protected Driver getDriver() throws SQLException {
        return loadDriver(DataSourceType.POSTGRESQL.getDriverClass());
    }

    @Override
    protected boolean filterSchema(String schemaName) {
        if (schemaName.equalsIgnoreCase("pg_catalog")) {
            return false;
        }
        return super.filterSchema(schemaName);
    }

    @Override
    protected String getDDLColumnName() {
        return "";
    }

    @Override
    public List<String> getDatabases() {
        try (Connection connection = this.getConnection();
             PreparedStatement preparedStatement = connection.prepareStatement(QUERY_DATABASES_SQL);
             ResultSet resultSet = preparedStatement.executeQuery()) {

            List<String> catalogs = Lists.newArrayList();
            while (resultSet.next()) {
                String catalogName = resultSet.getString(1);
                catalogs.add(catalogName);
            }
            return catalogs;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    public String getCreateTableScript(String databaseName, String tableName) {
        try (Connection connection = connectionFactory.openConnection();
             Statement statement = connection.createStatement()){
            statement.execute(create_table_function_sql);
            String tableSql = createTableSql(databaseName, tableName);
            System.out.println("createTableSql:" + tableSql);
            ResultSet dbResult = statement.executeQuery(tableSql);
            if (dbResult.next()) {
                return dbResult.getString(getDDLColumnName());
            } else {
                return "DDL is not available";
            }
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error", e);
        }
    }

    @Override
    protected String createTableSql(String schemaName, String tableName) {
        return "select generate_create_table_statement('"+ schemaName +"', '" + tableName + "') as ct_script;";
    }

    private static final String create_table_function_sql =
            "CREATE OR REPLACE FUNCTION generate_create_table_statement(p_schema_name varchar, p_table_name varchar)\n" +
            "  RETURNS text AS\n" +
            "$BODY$\n" +
            "DECLARE\n" +
            "    v_table_ddl   text;\n" +
            "    column_record record;\n" +
            "BEGIN\n" +
            "    FOR column_record IN \n" +
            "        SELECT \n" +
            "            b.nspname as schema_name,\n" +
            "            b.relname as table_name,\n" +
            "            a.attname as column_name,\n" +
            "            pg_catalog.format_type(a.atttypid, a.atttypmod) as column_type,\n" +
            "            CASE WHEN \n" +
            "                (SELECT substring(pg_catalog.pg_get_expr(d.adbin, d.adrelid) for 128)\n" +
            "                 FROM pg_catalog.pg_attrdef d\n" +
            "                 WHERE d.adrelid = a.attrelid AND d.adnum = a.attnum AND a.atthasdef) IS NOT NULL THEN\n" +
            "                'DEFAULT '|| (SELECT substring(pg_catalog.pg_get_expr(d.adbin, d.adrelid) for 128)\n" +
            "                              FROM pg_catalog.pg_attrdef d\n" +
            "                              WHERE d.adrelid = a.attrelid AND d.adnum = a.attnum AND a.atthasdef)\n" +
            "            ELSE\n" +
            "                ''\n" +
            "            END as column_default_value,\n" +
            "            CASE WHEN a.attnotnull = true THEN \n" +
            "                'NOT NULL'\n" +
            "            ELSE\n" +
            "                'NULL'\n" +
            "            END as column_not_null,\n" +
            "            a.attnum as attnum,\n" +
            "            e.max_attnum as max_attnum\n" +
            "        FROM \n" +
            "            pg_catalog.pg_attribute a\n" +
            "            INNER JOIN \n" +
            "             (SELECT c.oid,\n" +
            "                n.nspname,\n" +
            "                c.relname\n" +
            "              FROM pg_catalog.pg_class c\n" +
            "                   LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace\n" +
            "              WHERE c.relname ~ ('^('||p_table_name||')$') and n.nspname ~ ('^('||p_schema_name||')$')\n" +
            //"                AND pg_catalog.pg_table_is_visible(c.oid)\n" +
            "              ORDER BY 2, 3) b\n" +
            "            ON a.attrelid = b.oid\n" +
            "            INNER JOIN \n" +
            "             (SELECT \n" +
            "                  a.attrelid,\n" +
            "                  max(a.attnum) as max_attnum\n" +
            "              FROM pg_catalog.pg_attribute a\n" +
            "              WHERE a.attnum > 0 \n" +
            "                AND NOT a.attisdropped\n" +
            "              GROUP BY a.attrelid) e\n" +
            "            ON a.attrelid=e.attrelid\n" +
            "        WHERE a.attnum > 0 \n" +
            "          AND NOT a.attisdropped\n" +
            "        ORDER BY a.attnum\n" +
            "    LOOP\n" +
            "        IF column_record.attnum = 1 THEN\n" +
            "            v_table_ddl:='CREATE TABLE \"'||column_record.schema_name||'\".\"'||column_record.table_name||'\" (';\n" +
            "        ELSE\n" +
            "            v_table_ddl:=v_table_ddl||',';\n" +
            "        END IF;\n" +
            " \n" +
            "        IF column_record.attnum <= column_record.max_attnum THEN\n" +
            "            v_table_ddl:=v_table_ddl||chr(10)||\n" +
            "                     '    \"'||column_record.column_name||'\" '||column_record.column_type||' '||column_record.column_default_value||' '||column_record.column_not_null;\n" +
            "        END IF;\n" +
            "    END LOOP;\n" +
            " \n" +
            "    v_table_ddl:=v_table_ddl||');';\n" +
            "    RETURN v_table_ddl;\n" +
            "END;\n" +
            "$BODY$\n" +
            "  LANGUAGE 'plpgsql' COST 100.0 SECURITY INVOKER;";
}
